﻿using RockBus.Configuration.Interfaces;
using RockBus.DataContracts;
using RockBus.Exceptions;
using RockBus.Logging.Interfaces;
using RockBus.Messages.Interfaces;
using RockBus.Routing.Interfaces;
using RockBus.Subscription.Interfaces;
using RockBus.Utilities;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Threading;

namespace RockBus.Routing
{
    public class EvaluationManager : IMessageProcessor
    {
        public EvaluationManager(
            ILogger logger,
            IReadOnlyConfiguration configuration,
            IMessageFactory messageFactory,
            IQueueManager queueManager,
            IEvaluator<SubscriberConfiguration>[] subscriptionEvaluators)
        {
            this.Logger = logger;
            this.Configuration = configuration;
            this.MessageFactory = messageFactory;
            this.QueueManager = queueManager;
            this.SubscriptionEvaluators = subscriptionEvaluators;

            this.InitializeSubscriptionEvaluators(new SubscriberConfiguration());
            this.Configuration.OnChanged += this.Configuration_OnChanged;
        }

        private void Configuration_OnChanged(object sender, EventArgs e)
        {
            this.InitializeSubscriptionEvaluators(this.Configuration.SubscriberConfiguration);
        }

        private ILogger Logger { get; set; }

        private IReadOnlyConfiguration Configuration { get; set; }

        private IMessageFactory MessageFactory { get; set; }

        private IQueueManager QueueManager { get; set; }

        private IEvaluator<SubscriberConfiguration>[] SubscriptionEvaluators { get; set; }

        private SubscriberConfiguration SubscriberConfiguration { get; set; }

        private readonly ReaderWriterLock _reInitializationLockObject = new ReaderWriterLock();

        public MessageProcessorEnum ProcessingStep { get { return MessageProcessorEnum.EvaluationManager; } }

        public void ProcessMessage(IMessageContext messageContext)
        {
            this.EvaluateSubscriptions(messageContext);
            this.NormalizeSubscriptions(messageContext);
        }

        private void EvaluateSubscriptions(IMessageContext messageContext)
        {
            messageContext.EvaluationResult = new EvaluationResult(this.Logger);
            this.EvaluateSubscriptions(messageContext.RequestMessage, messageContext.EvaluationResult);
            messageContext.SetRoutingHeaders(messageContext.EvaluationResult.GetDeliveryInfos());
        }

        private void EvaluateSubscriptions(IMessage messageWrapper, EvaluationResult result)
        {
            try
            {
                this._reInitializationLockObject.AcquireReaderLock(this.Configuration.InternalConfiguration.ReinitLockTimeoutMs);
                this.EvaluateSubscriptionsInternal(messageWrapper, result);
            }
            finally
            {
                if (this._reInitializationLockObject.IsReaderLockHeld)
                {
                    this._reInitializationLockObject.ReleaseReaderLock();
                }
            }
        }

        private void EvaluateSubscriptionsInternal(IMessage messageWrapper, EvaluationResult result)
        {
            this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "Action {0};", messageWrapper.Action);

            foreach (var evaluator in this.SubscriptionEvaluators)
            {
                evaluator.Evaluate(messageWrapper, result);
            }
        }

        private void InitializeSubscriptionEvaluators(SubscriberConfiguration sc)
        {
            if (SubscriberConfiguration.IsEqual(this.SubscriberConfiguration, sc))
            {
                return;
            }

            this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "Reinitializing Evaluators.");
            try
            {
                this._reInitializationLockObject.AcquireWriterLock(this.Configuration.InternalConfiguration.ReinitLockTimeoutMs);

                this.SubscriberConfiguration = sc.Clone<SubscriberConfiguration>();

                foreach (var evaluator in this.SubscriptionEvaluators)
                {
                    evaluator.Initialize(this.SubscriberConfiguration);
                }
            }
            catch (Exception exc)
            {
                this.Logger.LogError(MethodBase.GetCurrentMethod(), exc.ToString());
            }
            finally
            {
                if (this._reInitializationLockObject.IsWriterLockHeld)
                {
                    this._reInitializationLockObject.ReleaseWriterLock();
                }
            }
        }

        private void NormalizeSubscriptions(IMessageContext messageContext)
        {
            if (messageContext.RoutingHeaders.IsNullOrEmpty())
            {
                var description = string.Format("No Subscriptions for {0}", messageContext);
                if (messageContext.ThrowOnNoSubscriptions)
                {
                    var routingException = new RoutingException(description);
                    messageContext.Exception = routingException;
                    this.QueueManager.ProcessMessage(messageContext);
                }
                else
                {
                    this.Logger.LogDebug(MethodBase.GetCurrentMethod(), description);
                    messageContext.ProcessingData.MessageDirection = MessageDirectionEnum.End;
                }

                this.QueueManager.ProcessMessage(messageContext);
                return;
            }

            // Split up into persistent and non-persistent messageContext
            //  The persistent one takes the Tcs, the non-persistent ones take the Tcs is there's no persistent one.
            //  The non-persistent ones each go their own way

            List<RoutingHeader> persistentRoutingHeaders = new List<RoutingHeader>();
            List<RoutingHeader> transientRoutingHeaders = new List<RoutingHeader>();

            // Split RoutingHeaders and group them by Persistent/nonpersistent
            //  and work out if a response can be produced
            bool containsPersistentRRSubscription = false;
            bool containsNonPersistentRRSubscription = false;
            foreach (var rh in messageContext.RoutingHeaders)
            {
                if (rh.SubscriptionDetail.Persistent)
                {
                    persistentRoutingHeaders.Add(rh);
                    if (rh.IsRequestResponse)
                    {
                        containsPersistentRRSubscription = true;
                    }
                }
                else
                {
                    // Override the delivery attempt parameters if they weren't set
                    if (!rh.SubscriptionDetail.DeliveryAttemptIntervalSecs.HasValue)
                    {
                        rh.SubscriptionDetail.DeliveryAttemptIntervalSecs = this.Configuration.InternalConfiguration.DefaultTransientDeliveryAttemptIntervalSecs;
                    }
                    if (!rh.SubscriptionDetail.MaxDeliveryAttemptCount.HasValue)
                    {
                        rh.SubscriptionDetail.MaxDeliveryAttemptCount = this.Configuration.InternalConfiguration.DefaultTransientMaxDeliveryAttemptCount;
                    }

                    transientRoutingHeaders.Add(rh);
                    if (rh.IsRequestResponse)
                    {
                        containsNonPersistentRRSubscription = true;
                    }
                }
            }

            // Determine who will own the TCS (and who will complete the publisher's request with any potential response)

            // Rip apart the incoming MessageContext
            var processingData = new ProcessingData(messageContext.ProcessingData);
            var requestMessage = messageContext.RequestMessage;
            var tcs = messageContext.PublisherTcs;
            var evaluationResult = messageContext.EvaluationResult;
            var receivePortInfo = messageContext.ReceivePortInfo;
            var transaction = messageContext.Transaction;

            bool tcsOwned = false;
            IMessageContext correlationMessageContext = null;
            IMessageContext persistentMessageContext = null;
            var transientMessageContexts = new List<IMessageContext>();
            switch (messageContext.ProcessingData.PublisherExchangeType)
            {
                case ExchangeTypeEnum.OneWay:
                case ExchangeTypeEnum.TwoWayAck:
                    if (persistentRoutingHeaders.Count > 0)
                    {
                        tcsOwned = true;
                        persistentMessageContext = this.MessageFactory.CreatePersistentMessageContext(
                            processingData,
                            requestMessage,
                            transaction,
                            persistentRoutingHeaders,
                            tcs,
                            evaluationResult,
                            receivePortInfo
                            );
                        persistentMessageContext.ProcessingData.MessageDirection = MessageDirectionEnum.InboundRequest;
                    }

                    foreach (var rh in transientRoutingHeaders)
                    {
                        var transientMessageContext = this.MessageFactory.CreateTransientMessageContext(
                            processingData,
                            requestMessage,
                            rh,
                            null,
                            null,
                            receivePortInfo
                        );
                        transientMessageContext.ProcessingData.MessageDirection = MessageDirectionEnum.OutboundRequest;
                        transientMessageContexts.Add(transientMessageContext);
                    }
                    //else
                    //{
                    //    this.Logger.LogWarning(MethodInfo.GetCurrentMethod(), "WARNING: No subscription(s); {0}", messageContext.ToString());
                    //    throw new RoutingException("No subscription(s)");
                    //}
                    break;

                case ExchangeTypeEnum.OneWayRequestResponse:
                case ExchangeTypeEnum.TwoWayAckRequestResponse:

                    if (containsPersistentRRSubscription)
                    {
                        correlationMessageContext = this.MessageFactory.CreatePersistentMessageContext(
                            processingData,
                            requestMessage,
                            transaction,
                            null,
                            null,
                            evaluationResult,
                            receivePortInfo
                            );
                        correlationMessageContext.ProcessingData.MessageDirection = MessageDirectionEnum.InboundRequest;
                    }

                    if (persistentRoutingHeaders.Count > 0)
                    {
                        persistentMessageContext = this.MessageFactory.CreatePersistentMessageContext(
                            processingData,
                            requestMessage,
                            transaction,
                            persistentRoutingHeaders,
                            null,
                            null,
                            receivePortInfo
                            );
                        persistentMessageContext.ProcessingData.MessageDirection = MessageDirectionEnum.InboundRequest;
                    }

                    foreach (var rh in transientRoutingHeaders)
                    {
                        var transientMessageContext = this.MessageFactory.CreateTransientMessageContext(
                            processingData,
                            requestMessage,
                            rh,
                            null,
                            null,
                            receivePortInfo
                        );
                        transientMessageContext.ProcessingData.MessageDirection = MessageDirectionEnum.OutboundRequest;
                        transientMessageContexts.Add(transientMessageContext);
                    }
                    //else
                    //{
                    //    this.Logger.LogWarning(MethodInfo.GetCurrentMethod(), "WARNING: No subscription(s) that could produce a response; {0}", messageContext.ToString());
                    //    throw new RoutingException("No subscription(s) that could produce a response");
                    //}
                    break;

                case ExchangeTypeEnum.TwoWayRequestResponse:
                    if (containsPersistentRRSubscription || containsNonPersistentRRSubscription)
                    {
                        tcsOwned = true;
                    }

                    if (containsPersistentRRSubscription)
                    {
                        correlationMessageContext = this.MessageFactory.CreatePersistentMessageContext(
                            processingData,
                            requestMessage,
                            transaction,
                            null,
                            tcs,
                            evaluationResult,
                            receivePortInfo
                            );
                        correlationMessageContext.ProcessingData.MessageDirection = MessageDirectionEnum.InboundRequest;
                    }

                    if (persistentRoutingHeaders.Count > 0)
                    {
                        persistentMessageContext = this.MessageFactory.CreatePersistentMessageContext(
                            processingData,
                            requestMessage,
                            transaction,
                            persistentRoutingHeaders,
                            null,
                            null,
                            receivePortInfo
                            );
                        persistentMessageContext.ProcessingData.MessageDirection = MessageDirectionEnum.InboundRequest;
                    }

                    foreach (var rh in transientRoutingHeaders)
                    {
                        var transientMessageContext = this.MessageFactory.CreateTransientMessageContext(
                            processingData,
                            requestMessage,
                            rh,
                            !containsPersistentRRSubscription && containsNonPersistentRRSubscription ? tcs : null,
                            !containsPersistentRRSubscription && containsNonPersistentRRSubscription ? evaluationResult : null,
                            receivePortInfo
                        );
                        transientMessageContext.ProcessingData.MessageDirection = MessageDirectionEnum.OutboundRequest;
                        transientMessageContexts.Add(transientMessageContext);
                    }
                    //else
                    //{
                    //    this.Logger.LogWarning(MethodInfo.GetCurrentMethod(), "WARNING: No subscription(s) that could produce a response; {0}", messageContext.ToString());
                    //    throw new RoutingException("No subscription(s) that could produce a response");
                    //}
                    break;

                default:
                    break;
            }

            // If none of the new MessageContext(s) owned the PublisherTcs, we complete it here
            if (!tcsOwned)
            {
                messageContext.ProcessingData.MessageDirection = MessageDirectionEnum.End;
                this.QueueManager.ProcessMessage(messageContext);
            }

            if (correlationMessageContext != null)
            {
                this.QueueManager.ProcessMessage(correlationMessageContext);
                //this.ProcessorDictionary[MessageProcessorEnum.CorrelationManager].ProcessMessage(correlationMessageContext);
            }

            if (persistentMessageContext != null)
            {
                this.QueueManager.ProcessMessage(persistentMessageContext);
            }

            foreach (var transientMessageContext in transientMessageContexts)
            {
                this.QueueManager.ProcessMessage(transientMessageContext);
            }
        }
    }
}